package com.xf.kafkatest.test;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class PartitionedProducer {

    // Kafka 配置常量
    private static final String BOOTSTRAP_SERVERS = "hadoop03:9092";
    private static final String TOPIC = "test-topic";

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        // 创建配置对象
        Properties props = new Properties();

        // 配置 Kafka 的服务端地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);

        // 配置 key 的序列化类
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 配置 value 的序列化类
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 自定义分区器类（关键：指定自定义分区逻辑）
        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.xf.kafkatest.test.MyPartitioner");

        // 设置 ACK 策略（确保消息发送成功）
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        // 创建 Kafka Producer 实例
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 发送 100 条记录
        for (int i = 0; i < 100; i++) {
            // 构造 ProducerRecord，key 为字符串，value 为数字
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, Integer.toString(i), "Message-" + i);

            // 同步发送：阻塞当前线程，直到收到服务端响应
            RecordMetadata recordMetadata = producer.send(record).get();

            // 获取元数据信息
            String topic = recordMetadata.topic();
            int partition = recordMetadata.partition();
            long offset = recordMetadata.offset();
            String message = recordMetadata.toString();

            // 打印发送结果
            System.out.println("message:[" + message + "] sent with topic:" + topic + "; partition:" + partition + "; offset:" + offset);
        }

        // 关闭生产者，释放资源
        producer.close();
    }
}
